package org.yonggan.shop.utils

import org.apache.spark.rdd.RDD
import org.yonggan.shop.domain.LogJson

/**
  * 实时计算工具
  */
object RTCalculateUtils {

  /**
    * 1 统计总销售金额 从今天凌晨到目前的销售金额（Redis）
    */
  def calculateTotalMoney(rdd: RDD[LogJson]): Unit = {

    val resRDD: RDD[(String, Double)] = rdd.map {
      lj =>
        val day = lj.time.split(" ")(0)
        val money = lj.total_money
        (day, money)
    }.reduceByKey(_ + _).cache()

    // 历史的总额度
    val resultArr: Array[(String, Double)] = resRDD.collect()

    val jedis = JedisPoolUtils.geJedis(8)
    //总金额
    val totalMoney = resultArr.map(_._2).sum
    jedis.incrByFloat("shop—total-money", totalMoney)

    // 存储每天数据
    resultArr.foreach(tp => {
      jedis.hincrByFloat("shop—rt-money", tp._1, tp._2)
    })
    // 释放连接
    jedis.close()
    // 释放缓存
    resRDD.unpersist(false)

  }


  /**
    * 2 按照分类进行累加销售金额（Redis）
    */
  def calculateCategoryTotalMoney(rdd: RDD[LogJson]): Unit = {

    val cidGoodsRDD: RDD[(String, Double)] = rdd.flatMap(_.goods.map {
      (g => (g.cid, g.money))
    }).reduceByKey(_ + _)

    // 存储每天分类数据
    cidGoodsRDD.foreachPartition(itor => {
      val jedis = JedisPoolUtils.geJedis(8)
      itor.foreach(tp => {
        jedis.hincrByFloat("shop—cid-rt-money", tp._1, tp._2)
      })
      // 释放连接
      jedis.close()
    })

    // 释放缓存
    cidGoodsRDD.unpersist(false)
  }

  /**
    *  3 按照省份进行累加，并且按照省份销售金额的从高到低进行排序（Redis）
    */
  def calculateProvinceTotalMoney(rdd: RDD[LogJson]): Unit = {

    val proRDD: RDD[(String, Double)] =
    rdd.map{
     lj=>
       (lj.province,lj.total_money)
    }.reduceByKey(_ + _)

    // 存储每天分类数据
    proRDD.foreachPartition(itor => {
      val jedis = JedisPoolUtils.geJedis(8)
      itor.foreach(tp => {
        // 使用redis中排序
        jedis.zincrby("shop-zone-data",tp._2,tp._1)
      })
      // 释放连接
      jedis.close()
    })

    // 释放缓存
    proRDD.unpersist(false)
  }

}
